package com.example.demo.thread.ch5.case13;

import com.example.demo.thread.ch4.case02.AbstractLogReader;
import com.example.demo.thread.ch4.case02.RecordSet;

import java.io.InputStream;
import java.util.concurrent.Exchanger;

/**
 * 使用Exchagner作为传输通道实例
 * @author: htc
 * @date: Created in 16:39 2019/1/14.
 */
public class ExchangerBasedLogReaderThread extends AbstractLogReader {

    private final Exchanger<RecordSet> exchanger;
    private volatile RecordSet nextToFill;
    private RecordSet consumedBatch;

    public ExchangerBasedLogReaderThread(InputStream in, int inputBufferSize, int batchSize) {
        super(in, inputBufferSize, batchSize);
        exchanger = new Exchanger<RecordSet>();
        nextToFill = new RecordSet(batchSize);
        consumedBatch = new RecordSet(batchSize);
    }

    @Override
    public RecordSet getNextToFill() {
        return nextToFill;
    }

    @Override
    protected RecordSet nextBatch() throws InterruptedException {
        consumedBatch = exchanger.exchange(consumedBatch);
        if (consumedBatch.isEmpty()){
            consumedBatch = null;
        }
        return consumedBatch;
    }

    @Override
    protected void publish(RecordSet recordBatch) throws InterruptedException {
        nextToFill = exchanger.exchange(recordBatch);
    }
}
